package com.taotao.job.executor.bootstrap;


import com.google.common.base.Preconditions;
import com.taotao.job.core.executor.ExecutorContext;
import com.taotao.job.core.util.JobBussinessUtil;
import com.taotao.job.executor.config.TaotaoExecutorConfig;
import com.taotao.job.executor.core.JobThread;
import com.taotao.job.executor.core.RunRateCache;
import com.taotao.job.executor.rpc.TaotaoJobExecutorRpcServer;
import com.taotao.job.rpc.core.utils.StringUtils;

import java.util.HashMap;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author fanbeibei
 * @date 2021/4/6 21:39
 */
public class TaotaoJobExecutor {
    /** -----------------------------------jobThread---------------------------------------------------------------------------------*/
    /**
     * jobId -> JobThread
     */
    protected static final ConcurrentHashMap<Long, JobThread> jobThreadCache = new ConcurrentHashMap<>();

    public static JobThread getJobThread(Long jobId) {
        Preconditions.checkArgument(null != jobId);
        return jobThreadCache.get(jobId);
    }

    public static JobThread runJob(ExecutorContext executorContext) {
        JobBussinessUtil.validExecutorContext(executorContext);

        Long jobId = executorContext.getJobId();
        JobThread jobThread = jobThreadCache.get(jobId);
        if (null == jobThread) {
            synchronized (jobThreadCache) {
                jobThread = jobThreadCache.get(jobId);
                if (null == jobThread) {
                    jobThread = new JobThread(jobId);
                    jobThreadCache.put(jobId, jobThread);
                    jobThread.start();
                }
            }
        }

        jobThread.addExecutorContext(executorContext);

        return jobThread;
    }

    /**
     * ------------------------------------运行进度缓存--------------------------------------------------------------------------------
     */
    protected static final RunRateCache runRateCache = RunRateCache.getInstance();

    public static void putRunRate(Long runInstanceId, String runRate) {
        Preconditions.checkArgument(null != runInstanceId);
        Preconditions.checkArgument(StringUtils.isNotBlank(runRate));

        runRateCache.putRunRate(runInstanceId, runRate);
    }

    public static String getRunRate(Long runInstanceId) {
        return runRateCache.getRunRate(runInstanceId);
    }

    public static HashMap<Long, String> getAllRunRate() {
        return runRateCache.getAllRunRate();
    }

    /**
     * --------------------------------------------------------------------------------------------------------------------
     */

    protected TaotaoExecutorConfig config;

    public TaotaoJobExecutor(TaotaoExecutorConfig config) {
        this.config = config;
    }

    protected TaotaoJobExecutorRpcServer taotaoJobRpcServer;


    public void init(){
        initRpcServer();
    }

    public void destroy(){

    }

    public void initRpcServer() {
        taotaoJobRpcServer = new TaotaoJobExecutorRpcServer(config.getPort());
        taotaoJobRpcServer.start();
    }
    /** --------------------------------------------------------------------------------------------------------------------*/




}
